package com.lenovo.lcdm.integration.config;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.ClassUtil;
import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONUtil;
import com.google.common.collect.Lists;
import com.lenovo.lcdm.integration.entity.EndpointConfig;
import com.lenovo.lcdm.integration.entity.EventLog;
import com.lenovo.lcdm.integration.entity.Message;
import com.lenovo.lcdm.integration.service.EventLogService;
import com.lenovo.lcdm.integration.service.MessageService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.core.io.ClassPathResource;

import java.io.File;
import java.io.IOException;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;

@Slf4j
public class KafkaConsumerContext {

    private final static Map<Long, EndpointConfig> endpointConfigMap = new ConcurrentHashMap<>();

    private final static Map<Long, KafkaConsumer<String, String>> consumerMap = new ConcurrentHashMap<>();

    private final static Map<Long, ScheduledFuture<?>> scheduleMap = new ConcurrentHashMap<>();

    private final static ScheduledExecutorService executorService = Executors.newScheduledThreadPool(50);

    public static void setEndpointConfigMap(EndpointConfig endpointConfig) {
        endpointConfigMap.put(endpointConfig.getId(), endpointConfig);
    }

    public static void startAllCosumerTask() {
        Set<Long> consumerIds = consumerMap.keySet();
        endpointConfigMap.keySet().stream()
                .filter(key -> !consumerIds.contains(key))
                .forEach(key -> {
                    createCosumer(endpointConfigMap.get(key));
                    addCosumerTask(key);
                });
    }

    public static void stopAllCosumerTask() {
        scheduleMap.keySet().forEach(KafkaConsumerContext::removeCosumerTask);
    }


    public static void addCosumerTask(Long id) {
        MessageService messageService = SpringUtil.getBean(MessageService.class);
        EventLogService eventLogService = SpringUtil.getBean(EventLogService.class);
        ScheduledFuture<?> scheduledFuture = executorService.scheduleAtFixedRate(() -> {
            KafkaConsumer<String, String> consumer = consumerMap.get(id);
//            log.info("topic: {}", endpointConfigMap.get(id).getTopic());
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            AtomicReference<List<Long>> messageIds = new AtomicReference<>(Lists.newArrayList());
            records.forEach(record -> {
                Message message = null;
                messageIds.set(Lists.newArrayList());
                try {
                    log.info("消费消息, topic :{}, data: {}", record.topic(), record.value());
                    message = new Message();
                    message.setObjectClass(endpointConfigMap.get(id).getMessageHandler());
                    message.setContent(record.value());
                    message.setCreator("lcdm");
                    message.setModifier("lcdm");
                    message.setCreateTime(DateTime.now());
                    message.setModifyTime(DateTime.now());
                    message.setDeleted(0);
                    messageService.save(message);
                    messageIds.get().add(message.getId());
                    ConsumerMessageHandler consumerMessageHandler = (ConsumerMessageHandler) ClassUtil.loadClass(endpointConfigMap.get(id).getMessageHandler()).getDeclaredConstructor().newInstance();
                    consumerMessageHandler.handleMessage(record.value());
                } catch (Exception e) {
                    EventLog eventLog = new EventLog();
                    eventLog.setMessageId(message.getId());
                    eventLog.setType("Consumer");
                    eventLog.setCategory(endpointConfigMap.get(id).getName());
                    eventLog.setRetryCount(0);
                    eventLog.setStatus("failed");
                    eventLog.setCreator("lcdm");
                    eventLog.setModifier("lcdm");
                    eventLog.setCreateTime(DateTime.now());
                    eventLog.setModifyTime(DateTime.now());
                    eventLog.setDeleted(0);
                    eventLogService.save(eventLog);
                }
            });
            consumer.commitAsync((offsets, exception) -> {
                if (exception != null) {
                    messageIds.get().forEach(messageId -> {
                        EventLog eventLog = new EventLog();
                        eventLog.setMessageId(messageId);
                        eventLog.setType("Consumer");
                        eventLog.setCategory(endpointConfigMap.get(id).getName());
                        eventLog.setRetryCount(0);
                        eventLog.setStatus("failed");
                        eventLog.setCreator("lcdm");
                        eventLog.setModifier("lcdm");
                        eventLog.setCreateTime(DateTime.now());
                        eventLog.setModifyTime(DateTime.now());
                        eventLog.setDeleted(0);
                        eventLogService.save(eventLog);
                    });
                }
            });
        }, 0, 1, TimeUnit.SECONDS);
        scheduleMap.put(id, scheduledFuture);
    }

    public static void removeCosumerTask(Long id) {
        scheduleMap.get(id).cancel(true);
        consumerMap.get(id).close();
        scheduleMap.remove(id);
        consumerMap.remove(id);
    }

    public static void createCosumer(EndpointConfig endpointConfig) {
        try {
            Map<String, Object> connectionInfo = JSONUtil.parseObj(endpointConfig.getConnectionInfo());
            connectionInfo.put("ssl.truststore.location", getSSlCertificatePath());
            endpointConfig.setConnectionInfo(connectionInfo);
            KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(connectionInfo);
            kafkaConsumer.subscribe(Arrays.stream(String.valueOf(connectionInfo.get("topic")).split(",")).toList());
            consumerMap.put(endpointConfig.getId(), kafkaConsumer);
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

    public static String getSSlCertificatePath() throws IOException {
        ClassPathResource classPathResource = new ClassPathResource("/jks/client_truststore.jks");
        String pathName = "/data/client_truststore.jks";
        File locationFile;
        if (!FileUtil.exist(pathName)) {
            locationFile = new File("/data/client_truststore.jks");
            FileUtils.copyInputStreamToFile(classPathResource.getInputStream(), locationFile);
        } else {
            locationFile = new File("/data/client_truststore.jks");
        }
        return locationFile.getAbsolutePath();
    }


}
